/**
 * Copyright © 2016-2021 The Thingsboard Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.example.iot.service.queue;

import com.example.iot.actors.ActorSystemContext;
import com.example.iot.msg.QueueToRuleEngineMsg;
import com.example.iot.msg.TbMsg;
import com.example.iot.queue.QueueConsumer;
import com.example.iot.queue.QueueFactory;
import com.example.iot.queue.TopicPartitionInfo;
import com.example.iot.queue.configuration.RuleEngineQueueConfig;
import com.example.iot.queue.message.RuleEngineQueueMsg;
import com.example.iot.queue.message.ProtoQueueMsg;
import com.example.iot.queue.processing.TbRuleEngineSubmitStrategy;
import com.example.iot.queue.processing.TbRuleEngineSubmitStrategyFactory;
import com.example.iot.queue.util.TopicPartitionService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.*;
import java.util.concurrent.*;

@Service
@Slf4j
public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<RuleEngineQueueMsg> {

    public static final String SUCCESSFUL_STATUS = "successful";
    public static final String FAILED_STATUS = "failed";
    @Value("${queue.rule-engine.poll-interval:25}")
    private long pollDuration;
    @Value("${queue.rule-engine.pack-processing-timeout:2000}")
    private long packProcessingTimeout;

    private final TbRuleEngineSubmitStrategyFactory submitStrategyFactory;
    //private final TbRuleEngineProcessingStrategyFactory processingStrategyFactory;
    private final QueueFactory queueFactory;
    private final RuleEngineQueueConfig ruleEngineQueueConfig;
    //private final TbRuleEngineDeviceRpcService tbDeviceRpcService;
    private QueueConsumer<ProtoQueueMsg<RuleEngineQueueMsg>> consumer;
    private ExecutorService submitExecutor;

    public DefaultTbRuleEngineConsumerService(TbRuleEngineSubmitStrategyFactory tbRuleEngineSubmitStrategyFactory,
                                              RuleEngineQueueConfig ruleEngineQueueConfig,
                                              QueueFactory queueFactory,
                                              ActorSystemContext actorContext) {
        super(actorContext, queueFactory.createRuleEngineConsumer());
        this.submitStrategyFactory = tbRuleEngineSubmitStrategyFactory;
        this.ruleEngineQueueConfig = ruleEngineQueueConfig;
        this.queueFactory = queueFactory;
        /*this.submitStrategyFactory = submitStrategyFactory;
        this.processingStrategyFactory = processingStrategyFactory;*/
        //this.tbDeviceRpcService = tbDeviceRpcService;
    }

    @PostConstruct
    public void init() {
        super.init("rule-engine-consumer", "rule-engine-notifications-consumer");
        consumer = queueFactory.createRuleEngineConsumer();
        submitExecutor = Executors.newSingleThreadExecutor();
    }

    @PreDestroy
    public void stop() {
        super.destroy();
        if (submitExecutor != null) {
            submitExecutor.shutdownNow();
        }
        //ruleEngineSettings.getQueues().forEach(config -> consumerConfigurations.put(config.getName(), config));
    }

    @Override
    @EventListener(ApplicationReadyEvent.class)
    @Order(value = 2)
    public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
        super.onApplicationEvent(applicationReadyEvent);
        Set<TopicPartitionInfo> tpiList = new HashSet<>();
        int size = ruleEngineQueueConfig.getPartitions();
        for (int i = 0; i < size; i++) {
            tpiList.add(TopicPartitionService.buildTopicPartitionInfo(ruleEngineQueueConfig.getTopic(), i));
        }
        consumer.subscribe(tpiList);
    }

    @Override
    protected void launchMainConsumers() {
        launchConsumer(consumer, ruleEngineQueueConfig);
    }

    @Override
    protected void stopMainConsumers() {
        consumer.unsubscribe();
    }

    private void launchConsumer(QueueConsumer<ProtoQueueMsg<RuleEngineQueueMsg>> consumer, RuleEngineQueueConfig ruleEngineQueueConfig) {
        consumersExecutor.execute(() -> {
            while (!stopped) {
                try {
                    List<ProtoQueueMsg<RuleEngineQueueMsg>> msgs = consumer.poll(pollDuration);
                    if (msgs.isEmpty()) {
                        continue;
                    }
                    TbRuleEngineSubmitStrategy submitStrategy = submitStrategyFactory.newInstance(ruleEngineQueueConfig.getTopic(), ruleEngineQueueConfig.getSubmitStrategy());
                    //TbRuleEngineProcessingStrategy ackStrategy = processingStrategyFactory.newInstance(configuration.getName(), configuration.getProcessingStrategy());

                    submitStrategy.init(msgs);

                    while (!stopped) {
                        //TbMsgPackProcessingContext ctx = new TbMsgPackProcessingContext(configuration.getName(), submitStrategy);
                        submitStrategy.submitAttempt((id, msg) -> submitExecutor.submit(() -> {
                            log.trace("[{}] Creating callback for message: {}", id, msg.getValue());
                            RuleEngineQueueMsg toRuleEngineMsg = msg.getValue();
                            //TenantId tenantId = new TenantId(new UUID(toRuleEngineMsg.getTenantIdMSB(), toRuleEngineMsg.getTenantIdLSB()));
                            /*TbMsgCallback callback = statsEnabled ?
                                    new TbMsgPackCallback(id, tenantId, ctx, stats.getTimer(tenantId, SUCCESSFUL_STATUS), stats.getTimer(tenantId, FAILED_STATUS)) :
                                    new TbMsgPackCallback(id, tenantId, ctx);*/
                            try {
                                forwardToRuleEngineActor(toRuleEngineMsg);
                                /*if (toRuleEngineMsg.getMsg() != null) {
                                } else {
                                    callback.onSuccess();
                                }*/
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }));
                        submitStrategy.stop();
                        break;

                        /*boolean timeout = false;
                        if (!ctx.await(configuration.getPackProcessingTimeout(), TimeUnit.MILLISECONDS)) {
                            timeout = true;
                        }

                        TbRuleEngineProcessingResult result = new TbRuleEngineProcessingResult(configuration.getName(), timeout, ctx);
                        if (timeout) {
                            printFirstOrAll(configuration, ctx, ctx.getPendingMap(), "Timeout");
                        }
                        if (!ctx.getFailedMap().isEmpty()) {
                            printFirstOrAll(configuration, ctx, ctx.getFailedMap(), "Failed");
                        }
                        ctx.printProfilerStats();

                        TbRuleEngineProcessingDecision decision = ackStrategy.analyze(result);
                        if (statsEnabled) {
                            stats.log(result, decision.isCommit());
                        }
                        if (decision.isCommit()) {
                            submitStrategy.stop();
                            break;
                        } else {
                            submitStrategy.update(decision.getReprocessMap());
                        }*/
                    }
                    consumer.commit();
                } catch (Exception e) {
                    if (!stopped) {
                        log.warn("Failed to process messages from queue.", e);
                        try {
                            Thread.sleep(pollDuration);
                        } catch (InterruptedException e2) {
                            log.trace("Failed to wait until the server has capacity to handle new requests", e2);
                        }
                    }
                }
            }
            log.info("TB Rule Engine Consumer stopped.");
        });
    }

    /*private void printFirstOrAll(TbRuleEngineQueueConfiguration configuration, TbMsgPackProcessingContext ctx, Map<UUID, TbProtoQueueMsg<ToRuleEngineMsg>> map, String prefix) {
        boolean printAll = log.isTraceEnabled();
        log.info("{} to process [{}] messages", prefix, map.size());
        for (Map.Entry<UUID, TbProtoQueueMsg<ToRuleEngineMsg>> pending : map.entrySet()) {
            ToRuleEngineMsg tmp = pending.getValue().getValue();
            TbMsg tmpMsg = TbMsg.fromBytes(configuration.getName(), tmp.getTbMsg().toByteArray(), TbMsgCallback.EMPTY);
            RuleNodeInfo ruleNodeInfo = ctx.getLastVisitedRuleNode(pending.getKey());
            if (printAll) {
                log.trace("[{}] {} to process message: {}, Last Rule Node: {}", new TenantId(new UUID(tmp.getTenantIdMSB(), tmp.getTenantIdLSB())), prefix, tmpMsg, ruleNodeInfo);
            } else {
                log.info("[{}] {} to process message: {}, Last Rule Node: {}", new TenantId(new UUID(tmp.getTenantIdMSB(), tmp.getTenantIdLSB())), prefix, tmpMsg, ruleNodeInfo);
                break;
            }
        }
    }

    @Override
    protected ServiceType getServiceType() {
        return ServiceType.TB_RULE_ENGINE;
    }

    @Override
    protected long getNotificationPollDuration() {
        return pollDuration;
    }

    @Override
    protected long getNotificationPackProcessingTimeout() {
        return packProcessingTimeout;
    }

    @Override
    protected void handleNotification(UUID id, TbProtoQueueMsg<ToRuleEngineNotificationMsg> msg, TbCallback callback) throws Exception {
        ToRuleEngineNotificationMsg nfMsg = msg.getValue();
        if (nfMsg.getComponentLifecycleMsg() != null && !nfMsg.getComponentLifecycleMsg().isEmpty()) {
            handleComponentLifecycleMsg(id, nfMsg.getComponentLifecycleMsg());
            callback.onSuccess();
        } else if (nfMsg.hasFromDeviceRpcResponse()) {
            TransportProtos.FromDeviceRPCResponseProto proto = nfMsg.getFromDeviceRpcResponse();
            RpcError error = proto.getError() > 0 ? RpcError.values()[proto.getError()] : null;
            FromDeviceRpcResponse response = new FromDeviceRpcResponse(new UUID(proto.getRequestIdMSB(), proto.getRequestIdLSB())
                    , proto.getResponse(), error);
            tbDeviceRpcService.processRpcResponseFromDevice(response);
            callback.onSuccess();
        } else {
            log.trace("Received notification with missing handler");
            callback.onSuccess();
        }
    }*/

    private void forwardToRuleEngineActor(RuleEngineQueueMsg toRuleEngineMsg) {
        TbMsg tbMsg = toRuleEngineMsg.getMsg();
        //TbMsg tbMsg = TbMsg.fromBytes(queueName, toRuleEngineMsg.getMsg().toByteArray(), callback);
        QueueToRuleEngineMsg msg;
        //ProtocolStringList relationTypesList = toRuleEngineMsg.getRelationTypesList();
        Set<String> relationTypes = null;
        /*if (relationTypesList != null) {
            if (relationTypesList.size() == 1) {
                relationTypes = Collections.singleton(relationTypesList.get(0));
            } else {
                relationTypes = new HashSet<>(relationTypesList);
            }
        }*/
        msg = new QueueToRuleEngineMsg(tbMsg, relationTypes, "Failure");
        actorContext.tell(msg);
    }

    /*@Scheduled(fixedDelayString = "${queue.rule-engine.stats.print-interval-ms}")
    public void printStats() {
        if (statsEnabled) {
            long ts = System.currentTimeMillis();
            consumerStats.forEach((queue, stats) -> {
                stats.printStats();
                statisticsService.reportQueueStats(ts, stats);
                stats.reset();
            });
        }
    }*/

}
